package com.briup.mr;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.KeyValueTextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import java.io.IOException;

/**
 * 专利引用次数统计
 *
 */
public class CiteCount  extends Configured implements Tool {

    public  static  class  CiteCountMapper extends  Mapper<Text,Text,Text,IntWritable>{
        private  final  IntWritable I=new IntWritable(1);
        @Override
        protected void map(Text key, Text value, Context context) throws IOException, InterruptedException {
            context.write(value,I);
        }
    }

    public  static  class CiteCountReducer extends Reducer<Text,IntWritable,Text,IntWritable>{
        @Override
        protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
            int  num=0;
            for (IntWritable value : values) {
               num+= value.get();
            }
            context.write(key,new IntWritable(num));
        }
    }


    @Override
    public int run(String[] args) throws Exception {
//        获取配置对象
        Configuration conf = getConf();
//        通过配置对象获取外部参数
        String in = conf.get("in");
        String out = conf.get("out");
//        设置切割参数
        conf.set("mapreduce.input.keyvaluelinerecordreader.key.value.separator",",");
//        构建job对象
        Job job = Job.getInstance(conf);
//        job的基础设置
        job.setJobName("citeCount");
        job.setJarByClass(this.getClass());
//        输入处理的设置以及输入路径的设置
        job.setInputFormatClass(KeyValueTextInputFormat.class);
        KeyValueTextInputFormat.addInputPath(job,new Path(in));
//        mapper类型的设置
        job.setMapperClass(CiteCountMapper.class);
//        map阶段的kv 类型设置
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);
//        设置reducer类型
        job.setReducerClass(CiteCountReducer.class);
//        设置reduce、阶段的kv输出类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);

//        设置输出处理器以及输出路径
        job.setOutputFormatClass(TextOutputFormat.class);
        TextOutputFormat.setOutputPath(job,new Path(out));
//        提交

        return job.waitForCompletion(true)?0:-1;
    }

    public static void main(String[] args) throws Exception {
        ToolRunner.run( new CiteCount(),args);
    }
}
